package com.tangji.signaltower.runners.client;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.http.DefaultHttpHeaders;
import io.netty.handler.codec.http.HttpClientCodec;
import io.netty.handler.codec.http.HttpObjectAggregator;
import io.netty.handler.codec.http.websocketx.TextWebSocketFrame;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshaker;
import io.netty.handler.codec.http.websocketx.WebSocketClientHandshakerFactory;
import io.netty.handler.codec.http.websocketx.WebSocketVersion;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.ssl.SslContext;
import io.netty.handler.ssl.SslContextBuilder;
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.SSLException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class ForexPoNettyClient {

    private static final Logger log = LoggerFactory.getLogger(ForexPoNettyClient.class);

    private static ForexPoNettyClient singleInstance = new ForexPoNettyClient();

    public static ForexPoNettyClient getInst() {
        return singleInstance;
    }

    private Channel channel;

    private boolean connectFlag = false;

    private String token;

    /**
     * 服务器ip地址
     */
//    private String hostUrl = "wss://socket.polygon.io/stocks";//股票
    private String hostUrl = "wss://ws.binaryws.com/websockets/v3?app_id=28054";//外汇
//    private String hostUrl = "wss://socket.polygon.io/crypto";//数字货币

    private String thirdPartyToken = "FMs_o_GOx7wGFbNH79A_SLM__xnhMBw_AYQ0h_";

    private EventLoopGroup eventLoopGroup;

    private WebSocketClientHandshaker handshaker;

    public Channel getChannel() {
        return channel;
    }

    public boolean isConnectFlag() {
        return connectFlag;
    }

    public ForexPoNettyClient setConnectFlag(boolean connectFlag) {
        this.connectFlag = connectFlag;
        return this;
    }

    public String getToken() {
        return token;
    }

    public ForexPoNettyClient setToken(String token) {
        this.token = token;
        return this;
    }

    public String getHostUrl() {
        return hostUrl;
    }

    public ForexPoNettyClient setHostUrl(String hostUrl) {
        this.hostUrl = hostUrl;
        return this;
    }

    public String getThirdPartyToken() {
        return thirdPartyToken;
    }

    public ForexPoNettyClient setThirdPartyToken(String thirdPartyToken) {
        this.thirdPartyToken = thirdPartyToken;
        return this;
    }

    public WebSocketClientHandshaker getHandshaker() {
        return handshaker;
    }

    public void setHandshaker(WebSocketClientHandshaker handshaker) {
        this.handshaker = handshaker;
    }

    public ForexPoNettyClient() {

    }

    private void reset() {
        this.handshaker = null;
        this.channel = null;
        this.connectFlag = false;
    }

    /**
     * 开始连接服务器
     *
     * @param token
     */
    public void startConnectServerAlways(String token) {
        while (true) {
            while (this.connectFlag == false) {
                //启动连接
                startConnectServer(token);
                try {
                    Thread.sleep(100);
                    log.error("ForexPoNettyClient,触发重新连接...");
                } catch (InterruptedException e) {
                }
            }

            try {
                Thread.sleep(1000);
                log.error("ForexPoNettyClient,socket主线程等待...");
            } catch (InterruptedException e) {
            }
        }
    }

    /**
     * 异常重连
     */
    private void startConnectServer(String t) {
        this.token = t == null ? this.token : t;
        //登录
        if (this.connectFlag) {
            log.info("ForexPoNettyClient,已经登录成功，不需要重复登录");
            return;
        }

        log.info("ForexPoNettyClient,开始连接Netty服务节点");

        this.reset();
        this.doConnect();
    }

    /**
     * 重连
     */
    private void doConnect() {
        try {
            URI uri = null;
            try {
                uri = new URI(this.hostUrl);
            } catch (URISyntaxException e) {
            }

            String scheme = uri.getScheme() == null ? "ws" : uri.getScheme();
            final String host = uri.getHost() == null ? "127.0.0.1" : uri.getHost();
            final int port;
            if (uri.getPort() == -1) {
                if ("ws".equalsIgnoreCase(scheme)) {
                    port = 80;
                } else if ("wss".equalsIgnoreCase(scheme)) {
                    port = 443;
                } else {
                    port = -1;
                }
            } else {
                port = uri.getPort();
            }

            if (!"ws".equalsIgnoreCase(scheme) && !"wss".equalsIgnoreCase(scheme)) {
                log.error("ForexPoNettyClient,Only WS(S) is supported.");
                return;
            }

            final boolean ssl = "wss".equalsIgnoreCase(scheme);
            final SslContext sslCtx;
            if (ssl) {
                try {
                    sslCtx = SslContextBuilder.forClient().trustManager(InsecureTrustManagerFactory.INSTANCE).build();
                } catch (SSLException e) {
                    log.error("ForexPoNettyClient,ssl build error", e);
                    return;
                }
            } else {
                sslCtx = null;
            }

            log.info("ForexPoNettyClient,连接远程数据源,host:{},port:{}", host, port);

            this.eventLoopGroup = new NioEventLoopGroup();
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(this.eventLoopGroup);
            bootstrap.channel(NioSocketChannel.class);
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
            bootstrap.option(ChannelOption.TCP_NODELAY, true);
            bootstrap.option(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT);

            bootstrap.handler(new LoggingHandler(LogLevel.INFO));

            ForexPoWebSocketClientHandler poWebSocketClientHandler = new ForexPoWebSocketClientHandler();
            bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel socketChannel) throws Exception {
                    ChannelPipeline p = socketChannel.pipeline();
                    if (sslCtx != null) {
                        p.addLast(sslCtx.newHandler(socketChannel.alloc(), host, port));
                    }
                    p.addLast(new HttpClientCodec());
                    p.addLast(new HttpObjectAggregator(1024 * 1024 * 10));
                    p.addLast(new IdleStateHandler(60, 10, 0, TimeUnit.SECONDS));
                    p.addLast(new ForexPoHeartBeatClientHandler());
                    /**
                     * 暂不使用
                     */
//                    p.addLast(poWebSocketClientHandler);

                }
            });

            ChannelFuture future = null;
            //防止初次链接不上 每隔十秒重连一次
            while (future == null || !future.isSuccess()) {
                try {
                    future = bootstrap.connect(host, port).addListener((ChannelFuture f) -> {
                                if (!f.isSuccess()) {
                                    log.info("ForexPoNettyClient,服务器连接失败!");
                                    this.connectFlag = false;
                                } else {
                                    this.connectFlag = true;
                                    log.info("ForexPoNettyClient,服务器连接成功!");
                                    this.channel = f.channel();
                                }
                            }
                    ).sync();
                } catch (Exception ex) {
                    log.error("ForexPoNettyClient,重连中,host:{},port:{} ......", host, port);
                    Thread.sleep(1000);
                }
            }

            //进行握手
            WebSocketClientHandshaker handshaker = WebSocketClientHandshakerFactory.newHandshaker(
                    uri, WebSocketVersion.V13, (String) null, true, new DefaultHttpHeaders());
            log.info("ForexPoNettyClient,准备握手...");

            poWebSocketClientHandler.setHandshaker(handshaker);
            handshaker.handshake(future.channel());
            //阻塞等待是否握手成功 通过5秒轮训来避免无限等待
//            handler.handshakeFuture().sync();
            int i = 0;
            while (!handshaker.isHandshakeComplete()) {
                if (i % 10 == 0) {
                    log.info("ForexPoNettyClient,握手中...");
                }
                if (i > 50) {
                    throw new RuntimeException("Handshake timeout!");
                }
                Thread.sleep(100);
                i++;
            }

            this.handshaker = handshaker;
            log.info("ForexPoNettyClient,握手成功");

//            JSONObject auth = new JSONObject();
//            auth.put("action", "auth");
//            auth.put("params", this.thirdPartyToken);
            List<String> ticks = new ArrayList<>();
            ticks.add("frxGBPUSD");
            ticks.add("frxEURUSD");
            Map<String, Object> request = new HashMap<>();
            request.put("ticks", ticks);
            request.put("subscribe", 1);

            String ticksStr = JSON.toJSONString(request);
//            future.channel().writeAndFlush(new TextWebSocketFrame(ticksStr));
            future.channel().writeAndFlush(new TextWebSocketFrame(ticksStr));

            future.channel().closeFuture().sync();
        } catch (Exception e) {
            log.error("ForexPoNettyClient,异常1", e);
        } finally {
            this.close();
        }
    }

    public void close() {
        if (eventLoopGroup != null) {
            eventLoopGroup.shutdownGracefully();
        }
        this.handshaker = null;
        this.channel = null;
        this.connectFlag = false;
        log.error("ForexPoNettyClient,数据源断开!");
    }

}
